package per.chao.mqtt.domain.protocol.connect;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.auth.Auth;
import per.chao.mqtt.domain.protocol.BaseProtocolHandler;
import per.chao.mqtt.domain.repo.PubRelMessageRepo;
import per.chao.mqtt.domain.repo.PublishMessageRepo;
import per.chao.mqtt.domain.repo.SessionRepo;
import per.chao.mqtt.domain.repo.SubscribeMessageRepo;
import per.chao.mqtt.domain.shared.store.DupPubRelMessageStore;
import per.chao.mqtt.domain.shared.store.DupPublishMessageStore;
import per.chao.mqtt.domain.shared.store.SessionStore;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:05
 **/
@Component
@Slf4j
public class Connect extends BaseProtocolHandler {
	@Autowired
	private SessionRepo sessionRepo;
	@Autowired
	private SubscribeMessageRepo subscribeRepo;
	@Autowired
	private PublishMessageRepo publishRepo;
	@Autowired
	private PubRelMessageRepo pubRelRepo;
	@Autowired
	private Auth auth;

	@Override
	public void handleMessage(Channel channel, Object msg) {
		MqttConnectMessage connectMsg = (MqttConnectMessage) msg;
		// 消息解码器异常
		if (connectMsg.decoderResult().isFailure()) {
			Throwable cause = connectMsg.decoderResult().cause();
			if (cause instanceof MqttUnacceptableProtocolVersionException) {
				MqttConnAckMessage refuseMqttConnAckMsg = getRefuseMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
				log.info("CONNECT - MqttUnacceptableProtocolVersionException");
				channel.writeAndFlush(refuseMqttConnAckMsg);
				channel.close();
				return;
			} else if (cause instanceof MqttIdentifierRejectedException) {
				MqttConnAckMessage refuseMqttConnAckMsg = getRefuseMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
				log.info("CONNECT - MqttIdentifierRejectedException");
				channel.writeAndFlush(refuseMqttConnAckMsg);
				channel.close();
				return;
			}
			channel.close();
			return;
		}

		// clientId为空或null的情况, 这里要求客户端必须提供clientId, 不管cleanSession是否为1, 此处没有参考标准协议实现
		if (StringUtils.isBlank(connectMsg.payload().clientIdentifier())) {
			MqttConnAckMessage refuseMqttConnAckMsg = getRefuseMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
			channel.writeAndFlush(refuseMqttConnAckMsg);
			channel.close();
			return;
		}

		// 用户名和密码验证, 这里要求客户端连接时必须提供用户名和密码, 不管是否设置用户名标志和密码标志为1, 此处没有参考标准协议实现
		String username = connectMsg.payload().userName();
		String password = connectMsg.payload().passwordInBytes() == null ? null : new String(connectMsg.payload().passwordInBytes(), CharsetUtil.UTF_8);
		if (!auth.valid(username, password)) {
			MqttConnAckMessage refuseMqttConnAckMsg = getRefuseMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
			log.info("CONNECT - CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD");
			channel.writeAndFlush(refuseMqttConnAckMsg);
			channel.close();
			return;
		}

		String clientIdentifier = connectMsg.payload().clientIdentifier();
		// 如果会话中已经存在这个clientId，就清除相关消息，关闭该clientId之前的连接
		if (sessionRepo.containsKey(connectMsg.payload().clientIdentifier())) {
			SessionStore sessionStore = sessionRepo.get(clientIdentifier);
			Channel previousChannel = sessionStore.getChannel();
			boolean cleanSession = sessionStore.isCleanSession();
			if (cleanSession) {
				sessionRepo.remove(clientIdentifier);
				subscribeRepo.remove(clientIdentifier);
				pubRelRepo.remove(clientIdentifier);
				publishRepo.remove(clientIdentifier);
			}
			previousChannel.close();
		}

		// 处理本次连接的遗嘱信息
		SessionStore sessionStore = new SessionStore();
		sessionStore.setClientId(clientIdentifier);
		sessionStore.setChannel(channel);
		sessionStore.setCleanSession(connectMsg.variableHeader().isCleanSession());
		sessionStore.setWillMessage(null);
		if (connectMsg.variableHeader().isWillFlag()) {
			MqttPublishMessage willMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
					new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(connectMsg.variableHeader().willQos()), connectMsg.variableHeader().isWillRetain(), 0),
					new MqttPublishVariableHeader(connectMsg.payload().willTopic(), 0),
					Unpooled.buffer().writeBytes(connectMsg.payload().willMessageInBytes())
			);
			sessionStore.setWillMessage(willMessage);
		}

		// 处理心跳包
		if (connectMsg.variableHeader().keepAliveTimeSeconds() > 0) {
			if (channel.pipeline().names().contains("idle")) {
				channel.pipeline().remove("idle");
			}
			channel.pipeline().addFirst("idle", new IdleStateHandler(0, 0, Math.round(connectMsg.variableHeader().keepAliveTimeSeconds() * 1.5f), TimeUnit.SECONDS));
		}

		// 至此存储会话信息及返回接收客户端连接
		sessionRepo.put(clientIdentifier, sessionStore);
		// 将clientId存储到channel的map中
		channel.attr(AttributeKey.valueOf("clientId")).set(clientIdentifier);
		// 客户端确认服务端是否持有会话，从而保持会话状态的一致
		boolean sessionPresent = sessionRepo.containsKey(clientIdentifier) && !connectMsg.variableHeader().isCleanSession();
		MqttConnAckMessage okResp = (MqttConnAckMessage) MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
				new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, sessionPresent),
				null
		);
		channel.writeAndFlush(okResp);
		log.info("CONNECT - clientId: {}, cleanSession: {}", clientIdentifier, connectMsg.variableHeader().isCleanSession());

		// 如果cleanSession为0, 需要重发同一clientId存储的未完成的QoS1和QoS2的DUP消息，为打开了cleanSession的同一clientId保存了相应的消息
		if (!connectMsg.variableHeader().isCleanSession()) {
			List<DupPublishMessageStore> dupPublishMsgs = publishRepo.getByClientId(clientIdentifier);
			List<DupPubRelMessageStore> dupPubRelMsgs = pubRelRepo.getByClientId(clientIdentifier);
			dupPublishMsgs.forEach(publishMsg -> {
				MqttPublishMessage dupPublishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.PUBLISH, true, MqttQoS.valueOf(publishMsg.getMqttQoS()), false, 0),
						new MqttPublishVariableHeader(publishMsg.getTopic(), publishMsg.getMessageId()),
						Unpooled.buffer().writeBytes(publishMsg.getMessageBytes())
				);
				channel.writeAndFlush(dupPublishMessage);
			});
			dupPubRelMsgs.forEach(dupPubRelMessage -> {
				MqttMessage dupPublishMessage = MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.PUBREL, true, MqttQoS.AT_LEAST_ONCE, false, 0),
						MqttMessageIdVariableHeader.from(dupPubRelMessage.getMessageId()),
						null
				);
				channel.writeAndFlush(dupPublishMessage);
			});
		}
	}

	/**
	 * 获取拒绝连接的ConnAck信息
	 *
	 * @param returnCode
	 * @return
	 */
	private MqttConnAckMessage getRefuseMqttConnAckMsg(MqttConnectReturnCode returnCode) {
		return (MqttConnAckMessage) MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
				new MqttConnAckVariableHeader(returnCode, false),
				null
		);
	}
}
